package com.atguigu.gmallrealtime.app.dwd.db;

import com.atguigu.gmallrealtime.common.Constant;
import com.atguigu.gmallrealtime.util.HBaseUtil;
import com.atguigu.gmallrealtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author yhm
 * @create 2023-09-27 14:52
 */
public class DwdTradeRefundPaySuc {
    public static void main(String[] args) {
        // TODO 1 创建flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5L));
        // TODO 2 添加检查点和状态后端
        //        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
        //
        //        //2.2 设置检查点超时时间
        //        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //        //2.3 设置job取消之后 检查点是否保留
        //        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //        //2.4 设置两个检查点之间最小的时间间隔
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //        //2.5 设置重启策略
        //        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        //
        //        env.setStateBackend(new HashMapStateBackend());
        //        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        //
        //        System.setProperty("HADOOP_USER_NAME","atguigu");

        // TODO 3 读取topic_db数据
        String groupId = "dwd_trade_refund_pay_suc";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));

        // TODO 4 筛选出退款成功数据
        Table rPayTable = tableEnv.sqlQuery("select \n" +
                "  data['id'] id,\n" +
                "  data['out_trade_no'] out_trade_no,\n" +
                "  data['order_id'] order_id,\n" +
                "  data['sku_id'] sku_id,\n" +
                "  data['payment_type'] payment_type,\n" +
                "  data['trade_no'] trade_no,\n" +
                "  data['total_amount'] total_amount,\n" +
                "  data['subject'] subject,\n" +
                "  data['refund_status'] refund_status,\n" +
                "  data['create_time'] create_time,\n" +
                "  data['callback_time'] callback_time,\n" +
                "  data['callback_content'] callback_content,\n" +
                "  data['operate_time'] operate_time,\n" +
                "  ts,\n" +
                "  proc_time\n" +
                "from topic_db\n" +
                "where `table`='refund_payment'\n" +
                "and `type`='update'\n" +
                "and data['refund_status']='1602'\n" +
                "and `old`['refund_status'] is not null");

        tableEnv.createTemporaryView("r_pay",rPayTable);

        // TODO 5 筛选退单明细表数据 update
        Table rInfoTable = tableEnv.sqlQuery("select \n" +
                "  data['user_id'] user_id, \n" +
                "  data['order_id'] order_id, \n" +
                "  data['sku_id'] sku_id, \n" +
                "  data['refund_type'] refund_type, \n" +
                "  data['refund_num'] refund_num, \n" +
                "  data['refund_amount'] refund_amount, \n" +
                "  data['refund_reason_type'] refund_reason_type, \n" +
                "  data['refund_reason_txt'] refund_reason_txt, \n" +
                "  data['refund_status'] refund_status, \n" +
                "  data['create_time'] create_time, \n" +
                "  ts,\n" +
                "  proc_time\n" +
                "from topic_db\n" +
                "where `table`='order_refund_info'\n" +
                "and `type`='update'\n" +
                "and data['refund_status'] = '0705'\n" +
                "and `old`['refund_status'] is not null");
        tableEnv.createTemporaryView("r_info",rInfoTable);

        // TODO 6 筛选订单表信息 update
        Table oiTable = tableEnv.sqlQuery("select\n" +
                "  data['id'] id,\n" +
                "  data['province_id'] province_id\n" +
                "from topic_db\n" +
                "where `table`='order_info'\n" +
                "and `type`='update'\n" +
                "and `data`['order_status'] = '1006'\n" +
                "and `old`['order_status'] is not null");

        tableEnv.createTemporaryView("oi",oiTable);

        // TODO 7 获取字典表数据
        tableEnv.executeSql(HBaseUtil.getBaseDicDDL());

        // TODO 8 join合并退款和退单表数据  lookup join 合并字典表数据
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "  r_pay.id,\n" +
                "  r_pay.out_trade_no,\n" +
                "  r_pay.order_id,\n" +
                "  r_pay.sku_id,\n" +
                "  r_pay.payment_type,\n" +
                "  base_dic.dic_name payment_type_name,\n" +
                "  r_pay.trade_no,\n" +
                "  r_pay.total_amount,\n" +
                "  r_pay.subject,\n" +
                "  r_pay.refund_status,\n" +
                "  r_pay.create_time,\n" +
                "  r_pay.callback_time,\n" +
                "  r_pay.callback_content,\n" +
                "  r_pay.operate_time,\n" +
                "  r_info.user_id,\n" +
                "  r_info.refund_type,\n" +
                "  r_info.refund_num,\n" +
                "  r_info.refund_amount,\n" +
                "  r_info.refund_reason_type,\n" +
                "  r_info.refund_reason_txt,\n" +
                "  province_id,\n" +
                "  r_pay.ts,\n" +
                "  r_pay.proc_time\n" +
                "from r_pay\n" +
                "join r_info \n" +
                "on r_pay.order_id = r_info.order_id\n" +
                "and r_pay.sku_id = r_info.sku_id\n" +
                "join oi\n" +
                "on r_pay.order_id = oi.id \n" +
                "left join base_dic  FOR SYSTEM_TIME AS OF r_pay.proc_time\n" +
                "on r_pay.payment_type = base_dic.rowkey");

        tableEnv.createTemporaryView("result_table",resultTable);
        // TODO 9 写出到kafka中
        tableEnv.executeSql("create table kafka_sink(\n" +
                "  id string,\n" +
                "  out_trade_no string,\n" +
                "  order_id string,\n" +
                "  sku_id string,\n" +
                "  payment_type string,\n" +
                "  payment_type_name string,\n" +
                "  trade_no string,\n" +
                "  total_amount string,\n" +
                "  subject string,\n" +
                "  refund_status string,\n" +
                "  create_time string,\n" +
                "  callback_time string,\n" +
                "  callback_content string,\n" +
                "  operate_time string,\n" +
                "  user_id string,\n" +
                "  refund_type string,\n" +
                "  refund_num string,\n" +
                "  refund_amount string,\n" +
                "  refund_reason_type string,\n" +
                "  refund_reason_txt string,\n" +
                "  province_id string,\n" +
                "  ts BIGINT,\n" +
                "  proc_time TIMESTAMP(3),\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ")"
           + MyKafkaUtil.getUpsertKafkaDLL(Constant.TOPIC_DWD_TRADE_REFUND_PAY_SUC));

        tableEnv.executeSql("insert into kafka_sink select * from result_table");

    }
}
